package util;

import annotation.NoSink;
import bean.TableProcess;
import bean.VisitorStats;
import com.alibaba.fastjson.JSONObject;
import common.Constant;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.hadoop.mapreduce.lib.fieldsel.FieldSelectionHelper;
import org.apache.kafka.clients.producer.ProducerRecord;
import sink.MyPhoenixSink;

import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;

public class GmallSinkUtil {

    public static void sendToKafka(DataStream<String> stream, String topic) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "hadoop162:9092,hadoop163:9092,hadoop164:9092");
        props.setProperty("transaction.timeout.ms", 15 * 60 * 1000 + ""); // broker要求事务超时时间不能超过15分钟


        stream.addSink(new FlinkKafkaProducer<String>(topic, new KafkaSerializationSchema<String>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(String element, @Nullable Long timestamp) {

                return new ProducerRecord<>(topic, element.getBytes(StandardCharsets.UTF_8));
            }
        }, props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE));
    }

    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "hadoop162:9092,hadoop163:9092,hadoop164:9092");
        props.setProperty("transaction.timeout.ms", 15 * 60 * 1000 + ""); // broker要求事务超时时间不能超过15分钟


        return new FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>>(
            "default",
            new KafkaSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> element, @Nullable Long timestamp) {
                    return new ProducerRecord<>(element.f1.getSinkTable(), element.f0.toString().getBytes(StandardCharsets.UTF_8));
                }
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getHbaseSink() {

        return new MyPhoenixSink();
    }

    public static <T> SinkFunction<T> getClickHouseSink(String db, String table, Class<T> tClass) {
        String driver = Constant.CLICKHOUSE_DRIVER;
        String url = Constant.CLICKHOUSE_URL_PRE + db;
        StringBuilder sb = new StringBuilder();
        // insert into table(a,b,c) values(?,?,?);
        sb
            .append("insert into ")
            .append(table + "(");

        // 各列和逗号拼接
        Field[] fields = tClass.getDeclaredFields();
        for (Field field : fields) {
            if (field.getAnnotation(NoSink.class) == null) {
                sb.append(field.getName() + ",");
            }
        }
        sb.deleteCharAt(sb.length() - 1); //删除最后一个字符
        sb.append(")values(");
        for (Field field : fields) {
            if (field.getAnnotation(NoSink.class) == null) {
                sb.append("?,");
            }
        }
        sb.deleteCharAt(sb.length() - 1); //删除最后一个字符
        //上面列数量的?和逗号拼接
        sb.append(")");

        return getJDBCSink(driver, url, sb.toString());
    }


    public static <T> SinkFunction<T> getJDBCSink(String driver, String url, String sql) {

        return JdbcSink.sink(sql,
                             new JdbcStatementBuilder<T>() {
                                 @Override
                                 public void accept(PreparedStatement ps, T t) throws SQLException {
                                     Field[] fields = t.getClass().getDeclaredFields();
                                     try {
                                         for (int i = 0, j = 1; i < fields.length; i++) {
                                             if (fields[i].getAnnotation(NoSink.class) == null) {
                                                 fields[i].setAccessible(true);
                                                 Object o = fields[i].get(t);//通过反射拿到属性值 :  属性.get(对象)
                                                 ps.setObject(j++, o);//这里不管属性有没有注解,i都会递增,所以要再定义一个变量,这个变量的变化应该只跟占位符有关
                                             }
                                         }
                                     } catch (IllegalAccessException e) {
                                         e.printStackTrace();
                                     }
                                 }
                             },
                             new JdbcExecutionOptions
                                 .Builder()
                                 .withBatchIntervalMs(1000)
                                 .withBatchSize(1024)
                                 .withMaxRetries(3)
                                 .build(),
                             new JdbcConnectionOptions
                                 .JdbcConnectionOptionsBuilder()
                                 .withDriverName(driver)
                                 .withUrl(url)
                                 .build());
    }
}
